package org.xxd.kafka.clients.producer.internals;

import org.xxd.kafka.clients.common.Cluster;
import org.xxd.kafka.clients.common.PartitionInfo;
import org.xxd.kafka.clients.producer.Partitioner;

import java.util.HashMap;
import java.util.List;
import java.util.Random;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author: XiaoDong.Xie
 * @create: 2020-09-27 11:32
 * @description:
 */
public class DefaultPartitioner implements Partitioner {

    private final AtomicInteger counter = new AtomicInteger(new Random().nextInt());


    @Override
    public int partition(String topic, String key, String value, Cluster cluster) {
        // 获取这个topic 的 partition 数量
        List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
        int numPartitions = partitions.size();
        if (key == null) {
            // 如果key是空的，那么就轮询
            // 这里就需要一个计数器，去一直轮询
            int nextValue = counter.getAndIncrement();
            List<PartitionInfo> availablePartitions = cluster.availablePartitionForTopic(topic);

            if (availablePartitions.size() > 0) {
                int part = DefaultPartitioner.toPositive(nextValue) % availablePartitions.size();
                return availablePartitions.get(part).partition();
            } else {
                return DefaultPartitioner.toPositive(nextValue) % numPartitions;
            }
        } else {
            // 如果key不是空的，就按key 的hashcode路由
            return key.hashCode() % numPartitions;
        }
    }

    private static int toPositive(int number) {
        return number & 0x7fffffff;
    }
}
